package com.qunar.toku.socket;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.jboss.netty.logging.InternalLogger;
import org.jboss.netty.logging.InternalLoggerFactory;
import org.jboss.netty.util.internal.LinkedTransferQueue;

import com.qunar.toku.configuration.SocketPoolProfile;
import com.qunar.toku.handler.ChannelClientHandler;
import com.qunar.toku.loadbalance.Server;
import com.qunar.toku.proxy.Future;

/**
 * SocketPool
 * 
 * @author zhangwan
 * 
 * @param <T>
 */
public class SocketPool<T> {
	private final InternalLogger logger = InternalLoggerFactory.getInstance(SocketPool.class);

	private final ChannelClientHandler handler;
	private final InetSocketAddress endPoint;
	private final CQueueL queue = new CQueueL();
	private final Object lock = new Object();
	private final SocketPoolProfile profile;

	public SocketPool(Class<T> clazz, SocketPoolProfile profile, Server server) {
		try {
			this.profile = profile;
			this.endPoint = new InetSocketAddress(server.getAddress(), server.getPort());
			this.handler = (ChannelClientHandler) clazz.newInstance();
		} catch (Exception e) {
			throw new InstantiationError(e.getMessage());
		}
	}

	public int count() {
		return queue.size();
	}

	private CSocket getSocket() throws Exception {
		CSocket socket = queue.dequeue();

		if (socket == null) {
			synchronized (lock) {
				if (queue.getTotal() < profile.getMaxPoolSize()) {
					CSocket rSocket = new CSocket(profile, endPoint, handler);
					queue.register(rSocket);
					socket = rSocket;
				}
			}
		}

		if (socket == null) {
			socket = queue.dequeue(profile.getWaitTimeout());
			if (socket == null) {
				logger.error("socket connection pool is full!");
				throw new TimeoutException("socket connection pool is full!");
			}
		}

		socket.setInPool(false);
		return socket;

	}

	public Future getFuture() throws Exception {
		CSocket socket = getSocket();
		Future future = null;

		if (!socket.getChannel().isOpen()) {
			destroy(socket);
			getFuture();
		} else {
			future = new Future(socket, handler.invokeMap, this);
			logger.info("get socket--------------------------------------current pool size:" + queue.size());
		}

		return future;
	}

	// 释放Socket到连接池
	public void release(CSocket socket) {
		if (socket == null) {
			logger.warn("socket is null when release(CSocket socket)");
		} else if (!socket.getChannel().isOpen() || !socket.connecting()) {
			logger.warn("socket is closed when release(CSocket socket)--" + socket.toString());
			destroy(socket);
		} else if (!socket.inPool()) {
			if (socket != null) {
				queue.enqueue(socket);
				socket.setInPool(true);
				logger.info("Socket:" + socket + ",release socket,current pool size:" + queue.size() + ",allSocket:" + queue.getTotal());
			}
		}
	}

	public void destroy() throws Exception {
		synchronized (this) {
			List<CSocket> csList = queue.getAllSocket();
			for (int i = 0; i < csList.size(); i++) {
				if (i < csList.size()) {
					CSocket socket = csList.get(i);
					destroy(socket);
				}
			}
		}
	}

	public void destroy(CSocket socket) {
		try {
			logger.warn("socket destroyed!--" + socket.toString());
			socket.disconnect();
		} catch (Throwable err) {
			logger.error("socket destroy error!--" + socket.toString(), err);
		} finally {
			queue.remove(socket);
		}
	}
}

class CQueueL extends LinkedTransferQueue<CSocket> {
	private static final long serialVersionUID = -9142257766208505704L;
	private CopyOnWriteArrayList<CSocket> _allSocket = new CopyOnWriteArrayList<CSocket>();

	public CSocket enqueue(CSocket element) {
		offer(element);
		return element;
	}

	public CSocket dequeue() {
		CSocket csocket = (CSocket) poll();
		return csocket;
	}

	public CSocket dequeue(long time) throws InterruptedException {
		CSocket csocket = (CSocket) poll(time, TimeUnit.MILLISECONDS);

		return csocket;
	}

	public void register(CSocket socket) {
		_allSocket.add(socket);
	}

	public synchronized boolean remove(CSocket socket) {
		_allSocket.remove(socket);
		return super.remove(socket);
	}

	public int getTotal() {
		return _allSocket.size();
	}

	public List<CSocket> getAllSocket() {
		return _allSocket;
	}

}
